package org.yuanzheng.function

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector
import org.yuanzheng.source.StationLog

/**
 * @author yuanzheng
 * @date 2020/6/17-22:12
 */
object TestProcessFunctionClass {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    import org.apache.flink.streaming.api.scala._
    //读取数据源
    val stream: DataStream[StationLog] = streamEnv.socketTextStream("192.168.1.9", 8888).map(line => {
      var split = line.split(",")
      new StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
    })
    //计算
    val result = stream.keyBy(_.callIn).process(new MonitorCallFail)
    result.print()
    streamEnv.execute()
  }
}

class MonitorCallFail extends KeyedProcessFunction[String, StationLog, String] {
  //使用一个状态对象记录时间
  lazy val timeState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("time", classOf[Long]))

  override def processElement(i: StationLog, context: KeyedProcessFunction[String, StationLog, String]#Context, collector: Collector[String]): Unit = {
    //从状态中获取时间
    var time = timeState.value()
    if (time == 0 && i.callType.equals("fail")) { //第一次失败
      //获取当前时间并注册定时器
      var nowTime = context.timerService().currentProcessingTime()
      //定时5秒后触发
      var onTime = nowTime + 5 * 1000L
      context.timerService().registerProcessingTimeTimer(onTime)
      //把当前时间保存到状态中
      timeState.update(onTime)
    }
    if (time != 0 && !i.callType.equals("fail")) {
      context.timerService().deleteProcessingTimeTimer(time)
      timeState.clear()
    }
  }

  //定时器
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, StationLog, String]#OnTimerContext, out: Collector[String]): Unit = {
    var warnStr = "触发时间：" + timestamp + ",手机号" + ctx.getCurrentKey
    out.collect(warnStr)
    timeState.clear()
  }
}